package com.its.common.kafka;

import cn.hutool.core.annotation.AnnotationUtil;
import cn.hutool.crypto.digest.DigestUtil;
import com.its.common.kafka.annotation.KafkaHandler;
import com.its.common.kafka.config.AsyncTaskQueue;
import com.its.common.kafka.config.KafkaConsumerProperties;
import com.its.common.kafka.config.KafkaMultiConsumerConfig;
import com.its.common.kafka.consumer.AbstractKafkaConsumerWorker;
import com.its.common.kafka.consumer.KafkaConsumerExecutor;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.core.Ordered;
import org.springframework.core.annotation.Order;
import org.springframework.util.StringUtils;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;


/**
 * @Author xiaxp
 * @Date 2021/11/17 22:13
 * @Description
 */
@Getter
@Setter
@Slf4j
@EnableConfigurationProperties({KafkaMultiConsumerConfig.class})
@ConditionalOnProperty(prefix = "its.kafka.consumer", name = "enabled", havingValue = "true")
@Order(Ordered.LOWEST_PRECEDENCE - 1)
public class KafkaConsumerConfigure implements CommandLineRunner, ApplicationContextAware {

    private static Map<String, KafkaConsumerExecutor> KAFKA_CONSUMER_EXEC_MAP = new ConcurrentHashMap<>();

    private ApplicationContext applicationContext;
    @Autowired
    private KafkaMultiConsumerConfig kafkaMultiConsumerConfig;

    @Override
    public void run(String... args) throws Exception {
        log.info("开始执行自定义Kafka消费者组");
        Map<String, KafkaConsumerProperties> consumerMap = extractKafkaConsumerFromAnnotation();
        List<KafkaConsumerProperties> configs = kafkaMultiConsumerConfig.getConfs();

        log.info("consumerMap.size() : " + consumerMap.size());
        log.info("configs.size() : " + configs.size());


        if(consumerMap.size() == 0 && (null == configs || configs.isEmpty())){
            log.warn("开启了Kafka消费者，但是配置的消费者处理器为0!");
            return;
        }

        Set<String> clazzSet = new HashSet<>();

        if(consumerMap.size() > 0) {
            consumerMap.forEach((key, prop) -> {
                Integer workerNum = prop.getWorkerNum();

                KafkaConsumerExecutor consumers = new KafkaConsumerExecutor
                        (prop.buildProperties(), prop.getTopic(), prop.getWorkerClaz(), workerNum);

                AsyncTaskQueue.handle(()-> consumers.execute());

                KAFKA_CONSUMER_EXEC_MAP.put(key, consumers);
                String clazz = prop.getWorkerClass();
                clazzSet.add(clazz);

                log.info("成功初始化消费者组：" + clazz + ", worker数量：" + workerNum
                        + ", bootstrapServers:" + prop.getBootstrapServers()
                        + ", topic:" + prop.getTopic()
                        + ", groupId:" + prop.getGroupId()
                        + ", className:" + prop.getWorkerClass()
                );
            });
        }

        for (KafkaConsumerProperties prop : configs) {
            Properties properties = prop.buildProperties();
            String topic = prop.getTopic();
            String bootstrapServers = prop.getBootstrapServers();
            String groupId = prop.getGroupId();

            if(!StringUtils.hasText(bootstrapServers)){
                throw new RuntimeException("<<<配置文件中>>> Kafka消费者处理器KafkaHandler必须包含bootstrapServers，且不能为空");
            }
            if(!StringUtils.hasText(topic)){
                throw new RuntimeException("<<<配置文件中>>> Kafka消费者处理器KafkaHandler必须包含topic，且不能为空");
            }
            if(!StringUtils.hasText(groupId)) {
                groupId = DigestUtil.md5Hex(bootstrapServers+"_"+topic);
                log.warn("<<<配置文件中>>> Kafka消费者处理器 " + prop.getWorkerClaz()
                        + "不包含groupId信息，为你创建groupId: " + groupId);
            }
            String key = bootstrapServers+"_"+topic+"_"+groupId;


            if(consumerMap.containsKey(key)){
                log.error("<<<配置文件中>>> Kafka消费者已经在包含了在：" + prop.getWorkerClass() +
                        ", 具有相同的bootstrapServers,topic,groupId");
                continue;
            }
            for (String claz : clazzSet) {
                if(Objects.equals(claz, prop.getWorkerClass())) {
                    log.warn("<<<配置文件中>>> Kafka消费者已经在包含了在：" + prop.getWorkerClass() +
                            ", 具有相同的bootstrapServers,topic, 但 groupId不同, 请确定是否需要两个不同的消费者组");
                }
            }
            KafkaConsumerExecutor consumers = new KafkaConsumerExecutor
                    (properties, topic, prop.getWorkerClaz(), prop.getWorkerNum());

            AsyncTaskQueue.handle(() -> consumers.execute());

            KAFKA_CONSUMER_EXEC_MAP.put(key, consumers);

            log.info("<<<配置文件中>>> 成功初始化消费者组："+prop.getWorkerClass() + ", worker数量：" + prop.getWorkerNum()
                    + ", bootstrapServers:" + prop.getBootstrapServers()
                    + ", topic:" + prop.getTopic()
                    + ", groupId:" + prop.getGroupId()
            );
        }

        log.info("结束定义Kafka消费者组，Kafka消费者开始工作...");

        addShutDownGracefully();
    }

    /**
     * 优雅关闭
     */
    private void addShutDownGracefully() {
        Runtime.getRuntime().addShutdownHook(new Thread(() -> KAFKA_CONSUMER_EXEC_MAP.forEach((k, exec)->{
            exec.shutdown();
            try {
                Thread.sleep(500L);
            } catch (InterruptedException e) {
            }
        })));
    }

    private Map<String, KafkaConsumerProperties> extractKafkaConsumerFromAnnotation() {
        Map<String, KafkaConsumerProperties> map = new HashMap<>(8);

        String[] KafkaHandlerBeanNames = applicationContext.getBeanNamesForAnnotation(KafkaHandler.class);

        if(KafkaHandlerBeanNames.length > 0){
            for (String kafkaHandlerBeanName : KafkaHandlerBeanNames) {
                Object kafkaHandlerBean = applicationContext.getBean(kafkaHandlerBeanName);
                Class<? extends Object> kafkaHandlerBeanClass = kafkaHandlerBean.getClass();;
                if(!AbstractKafkaConsumerWorker.class.isAssignableFrom(kafkaHandlerBeanClass)){
                    throw new RuntimeException("Kafka消费者必须继承AbstractKafkaConsumerWorker");
                }else{
                    Map<String, Object> annotationValueMap =
                            AnnotationUtil.getAnnotationValueMap(kafkaHandlerBeanClass, KafkaHandler.class);

                    if(null == annotationValueMap){
                        continue;
                    }
                    String bootstrapServers = (String) annotationValueMap.get("bootstrapServers");
                    String topic = (String) annotationValueMap.get("topic");
                    String groupId = (String) annotationValueMap.get("groupId");
                    Object workNum = annotationValueMap.get("workNum");

                    if(!StringUtils.hasText(bootstrapServers)){
                        throw new RuntimeException("Kafka消费者处理器KafkaHandler必须包含bootstrapServers，且不能为空");
                    }
                    if(!StringUtils.hasText(topic)){
                        throw new RuntimeException("Kafka消费者处理器KafkaHandler必须包含topic，且不能为空");
                    }
                    if(!StringUtils.hasText(groupId)) {
                        groupId = DigestUtil.md5Hex(bootstrapServers+"_"+topic);
                        log.warn("Kafka消费者处理器 " + kafkaHandlerBeanClass.getName()
                                + "不包含groupId信息，为你创建groupId(MD5摘要加密): " + groupId);
                    }
                    if(null == workNum || !(workNum instanceof Integer)){
                        workNum = 4;
                        log.warn("Kafka消费者处理器 " + kafkaHandlerBeanClass.getName()
                                + "默认为你创建worker数量: " + workNum);
                    }
                    String securityProtocol = (String) annotationValueMap.get("securityProtocol");
                    String mechanism = (String) annotationValueMap.get("mechanism");
                    String username = (String) annotationValueMap.get("username");
                    String password = (String) annotationValueMap.get("password");

                    KafkaConsumerProperties properties = KafkaConsumerProperties.builder()
                            .bootstrapServers(bootstrapServers)
                            .groupId(groupId)
                            .topic(topic)
                            .securityProtocol(securityProtocol)
                            .mechanism(mechanism)
                            .username(username)
                            .password(password)
                            .workerNum((Integer) workNum)
                            .workerClass(kafkaHandlerBeanClass.getName())
                            .build();
                    String key = bootstrapServers+"_"+topic+"_"+groupId;
                    map.put(key, properties);
                }
            }
        }
        return map;
    }



    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
}
